package server

import (
	"fmt"
	"github.com/golang/protobuf/proto"
	"go.uber.org/zap"
	"gse/protos"
	"gse/transport"
	"reflect"
	"runtime/debug"
	"sync"
	"sync/atomic"
	"time"
)

const DefaultSessionReqProcQueueSize = 120

// DefaultSessionReqWaitTimeout in millisecond
const DefaultSessionReqWaitTimeout = 1000

// DefaultReqProcessTimeout in millisecond
const DefaultReqProcessTimeout = 5000

// RequestCallback is the callback that is to process the remote response.
type RequestCallback func(ret proto.Message, err error)

//  OnRequest will be called when a remote request arrives. When returns an error the request will not pass to any handlers,
//	and the error will be wrapped into the RpcError message with RpcResponseStatusServiceProcessError status. If rsp is not nil ,
// then it will be used as the response to remote, and further request handler will not be invoked. If both rsp and err are nil
// the process will continue normally.
//	ses is the current session in communication.
//	param is the param for the requested service/request handler.
type OnRequest func(req *Request, param proto.Message, ses *Session, args ...interface{}) (rsp proto.Message, err error)

// AfterRequest is to be called after the response being sent to the remote caller.
// if the err param is not nil, it indicates that sending of the response is unsuccessful.
type AfterRequest func(req *Request, retM proto.Message, ses *Session, err error, args ...interface{}) error

// OnClose is to be called before closing the session
type OnClose func(ses *Session, args ...interface{})

// AfterCreate is to be called when the session is created.
type AfterCreate func(ses *Session, args ...interface{})

// AfterClose will be called immediately when the session is closed.
type AfterClose func(ses *Session, args ...interface{})

type HookReg struct {
	ExtraArgs []interface{}
	Fun       interface{}
}
type SessionConfig struct {
	// ReqWaitTimeout timeout for remote response.
	ReqWaitTimeout time.Duration
	// ReqProcessTimeout timeout for processing a remote request. if is zero, then no timeout will be fired.
	ReqProcessTimeout time.Duration
	OnRequest         []*HookReg
	// AfterRequest will be called after the response being sent to the remote caller.
	// if the err param is not nil, it indicates that sending of the response is unsuccessful.
	AfterRequest []*HookReg

	OnClose []*HookReg
	// AfterClose will be called immediately when the session is closed.
	AfterClose []*HookReg

	// AfterCreate will be invoked after successfully creating of a session.
	AfterCreate []*HookReg
}

func (scfg *SessionConfig) RegisterHook(hook interface{}, extraArgs ...interface{}) error {
	switch t := hook.(type) {
	default:
		return fmt.Errorf("unknown hook type %#v", hook)
	case AfterCreate:
		scfg.AfterCreate = append(scfg.AfterCreate, &HookReg{ExtraArgs: extraArgs, Fun: t})
	case OnClose:
		scfg.OnClose = append(scfg.OnClose, &HookReg{ExtraArgs: extraArgs, Fun: t})
	case AfterClose:
		scfg.AfterClose = append(scfg.AfterClose, &HookReg{ExtraArgs: extraArgs, Fun: t})
	case AfterRequest:
		scfg.AfterRequest = append(scfg.AfterRequest, &HookReg{ExtraArgs: extraArgs, Fun: t})
	case OnRequest:
		scfg.OnRequest = append(scfg.OnRequest, &HookReg{ExtraArgs: extraArgs, Fun: t})
	}
	return nil
}

type Session struct {
	cfg                  *SessionConfig
	conn                 transport.Connection
	transSequence        uint32
	transSequenceLock    *sync.Mutex
	rspContainer         *sync.Map
	serviceHandlerMap    **sync.Map
	reqProcQuota         chan bool
	reqProcTimerPool     chan *time.Timer
	nextEventSubSeriesID uint32
	// map[seriesID]*SubscriptionReg
	subscriptionRegs *sync.Map
}

type SubscriptionReg struct {
	seriesId       uint8
	eventProtoType proto.Message
	eventType      reflect.Type
	eventChan      chan *Push
}

func (sr *SubscriptionReg) EventChan() <-chan *Push {
	return sr.eventChan
}

func NewSession(conn transport.Connection, cfg *SessionConfig, serviceHandlerMap **sync.Map) *Session {
	s := &Session{
		cfg:               cfg,
		conn:              conn,
		transSequenceLock: new(sync.Mutex),
		rspContainer:      new(sync.Map),
		reqProcQuota:      make(chan bool, DefaultSessionReqProcQueueSize),
		reqProcTimerPool:  make(chan *time.Timer, DefaultSessionReqProcQueueSize),
		serviceHandlerMap: serviceHandlerMap,
		subscriptionRegs:  new(sync.Map),
	}

	for _, h := range cfg.AfterCreate {
		h.Fun.(AfterCreate)(s, h.ExtraArgs...)
	}
	return s
}

func (s *Session) GetRemoteAddr() string {
	return s.conn.RemoteAddr().String()
}

func (s *Session) GetLocalAddr() string {
	return s.conn.LocalAddr().String()
}

func (s *Session) getReqProcTimer() *time.Timer {
	select {
	case t := <-s.reqProcTimerPool:
		return t
	default:
		t := time.NewTimer(0)
		<-t.C
		return t
	}
}

func (s *Session) releaseReqProcTimer(t *time.Timer) {
	t.Reset(0)
	<-t.C
	select {
	case s.reqProcTimerPool <- t:
	default:
		t = nil
	}
}

// ReceiveEventPush receives the pushes that subscribed by SubEvent.
func (s *Session) ReceiveEventPush(p *Push) {
	if le := logger().Check(zap.DebugLevel, "received push"); le != nil {
		le.Write(zap.String("push", fmt.Sprintf("%+v", *p)))
	}

	var valid bool
	s.subscriptionRegs.Range(func(seriesID, r interface{}) bool {
		if p.seriesID == seriesID.(uint8) {
			valid = true
			reg := r.(*SubscriptionReg)
			if err := p.Decode(reg.eventType); err != nil {
				logger().Warn("failed to decode push data",
					zap.Error(err),
					zap.Uint8("series_id", p.seriesID),
					zap.Uint16("seq_id", p.seqID),
				)
				return false
			}
			reg.eventChan <- p
			return false
		}
		return true
	})
	if !valid {
		logger().Warn(
			"pushed event has not be subscribed, data dropped!",
			zap.String("push", fmt.Sprintf("%+v", *p)),
		)
	}
}

func (s *Session) SubEvent(eventName string, retMsgType proto.Message) (*SubscriptionReg, error) {
	req := new(protos.SubEvent)
	req.Name = eventName
	retM := new(protos.SubEventResponse)
	err := s.Request("#sub", req, retM, nil)
	if err != nil {
		return nil, err
	}

	switch retM.Status {
	case protos.EventRpcStatusTypes_OK:
		sr := &SubscriptionReg{
			seriesId:       uint8(retM.SeriesID),
			eventProtoType: retMsgType,
			eventChan:      make(chan *Push, 20),
			eventType:      reflect.TypeOf(reflect.ValueOf(retMsgType).Elem().Interface()),
		}

		s.subscriptionRegs.LoadOrStore(sr.seriesId, sr)
		return sr, nil
	default:
		return nil, fmt.Errorf("Failed to sub event '%v', reason: %v", eventName, retM.Msg)
	}
}

func (s *Session) genSubSeriesID() uint32 {
	id := atomic.AddUint32(&s.nextEventSubSeriesID, 1)
	if id > 254 {
		panic("event subscription generation exceeded 254.")
	}
	return id - 1
}

func (s *Session) Close() {
	if le := logger().Check(zap.DebugLevel, "closing session."); le != nil {
		le.Write(
			zap.String("remoteAddr", s.conn.RemoteAddr().String()),
		)
	}

	for _, h := range s.cfg.OnClose {
		if le := logger().Check(zap.DebugLevel, "calling OnClose hook"); le != nil {
			le.Write(
				zap.String("hookName", reflect.TypeOf(h).Name()),
				zap.String("remoteAddr", s.conn.RemoteAddr().String()),
			)
		}
		h.Fun.(OnClose)(s, h.ExtraArgs...)
	}
	s.conn.Close()
	for _, h := range s.cfg.AfterClose {
		if le := logger().Check(zap.DebugLevel, "calling AfterClose hook"); le != nil {
			le.Write(
				zap.String("hookName", reflect.TypeOf(h).Name()),
				zap.String("remoteAddr", s.conn.RemoteAddr().String()),
			)
		}
		h.Fun.(AfterClose)(s, h.ExtraArgs...)
	}
}

// TODO: generate transaction ID may be duplicated in concurrency on theory.
// transactionID generated in the server side is even.
func (s *Session) genTransactionID() uint16 {
	id := atomic.AddUint32(&s.transSequence, 2)
	if id > 40000 {
		s.transSequenceLock.Lock()
		id = atomic.LoadUint32(&s.transSequence)
		if id < 40000 {
			// in case other routine also  reached the threshold and entered the reset logic.
			id = atomic.AddUint32(&s.transSequence, 2)
		} else {
			atomic.StoreUint32(&s.transSequence, 2)
			id = 2
		}
		s.transSequenceLock.Unlock()
	}
	return uint16(id)
}

// Request send a request to the remote, and get response synchronously/asynchronously.
// the argument type of the RequestCallback is the same of retM
func (s *Session) Request(requestName string, m proto.Message, retM proto.Message, callback RequestCallback) error {
	mb, err := proto.Marshal(m)
	if err != nil {
		return err
	}
	req := NewRequest(requestName, s.genTransactionID(), mb)
	logger().Debug("sending request to remote.",
		zap.Uint16("trans_id", req.transactionID),
		zap.String("req_name", requestName),
		zap.String("remote_addr", s.conn.RemoteAddr().String()))
	reChan := make(chan *Response, 1)
	s.rspContainer.Store(req.transactionID, reChan)
	_, err = s.conn.SendData(req.Encode())
	if err != nil {
		s.rspContainer.Delete(req.transactionID)
		return err
	}

	if callback == nil {
		defer s.rspContainer.Delete(req.transactionID)
		select {
		case rsp := <-reChan:
			if rsp.GetStatus() != RpcResponseStatusOk {
				if rsp.GetStatus() == RpcResponseStatusServiceProcessError {
					errMsg := new(protos.RpcError)
					err = proto.Unmarshal(rsp.GetMessageData(), errMsg)
					if err != nil {
						return fmt.Errorf("Tansaction ID[%v], failed to decode RpcErr msg: %v", req.transactionID, err)
					}
					return fmt.Errorf("Tansaction ID[%v], rpc process error[%v]: %v", req.transactionID, errMsg.Code, errMsg.Msg)
				}
				return rsp.GetStatus()
			}
			err = proto.Unmarshal(rsp.GetMessageData(), retM)
			if err != nil {
				return fmt.Errorf("Tansaction ID[%v], failed to decode response msg: %v", req.transactionID, err)
			}
			return nil
		case <-time.After(s.cfg.ReqWaitTimeout):
			if le := logger().Check(zap.DebugLevel, "Request timed out."); le != nil {
				le.Write(
					zap.Uint16("trans_id", req.transactionID),
					zap.String("req_name", requestName),
					zap.String("remote_addr", s.conn.RemoteAddr().String()),
				)
			}
			return RequestWaitTiemedOut
		}
	}

	// Async request
	go func() {
		defer func() {
			s.rspContainer.Delete(req.transactionID)
			errI := recover()
			if errI != nil {
				logger().Warn("Request callback panic: %v", zap.Error(errI.(error)))
			}
		}()
		select {
		case rsp := <-reChan:
			if rsp.GetStatus() != RpcResponseStatusOk {
				if rsp.GetStatus() == RpcResponseStatusServiceProcessError {
					errMsg := new(protos.RpcError)
					err = proto.Unmarshal(rsp.GetMessageData(), errMsg)
					if err != nil {
						callback(nil, fmt.Errorf("Failed to decode RpcErr msg: %v", err))
					}
					callback(nil, fmt.Errorf("Rpc Process error[%v]: %v", errMsg.Code, errMsg.Msg))
				}
				callback(nil, rsp.GetStatus())
			}
			err = proto.Unmarshal(rsp.GetMessageData(), retM)
			if err != nil {
				callback(nil, fmt.Errorf("Failed to decode response msg: %v", err))
			}
			callback(retM, nil)
		case <-time.After(s.cfg.ReqWaitTimeout):
			callback(nil, RequestWaitTiemedOut)
		}
	}()
	return nil
}

func (s *Session) GetServiceHandlerInfo(serviceName string) (*ServiceHandlerRegisterInfo, error) {
	i, o := (*s.serviceHandlerMap).Load(serviceName)
	if !o {
		return nil, fmt.Errorf("service [%v] handler not registered!", serviceName)
	}
	return i.(*ServiceHandlerRegisterInfo), nil
}

func (s *Session) HandleRequest(req *Request) (err error) {
	var timer *time.Timer
	timer = s.getReqProcTimer()
	timer.Reset(s.cfg.ReqProcessTimeout)
	defer func() {
		s.releaseReqProcTimer(timer)
	}()
	if s.cfg.ReqProcessTimeout > 0 {
		select {
		case s.reqProcQuota <- true:
		case <-timer.C:
			return RpcResponseStatusTimeout
		}
	} else {
		s.reqProcQuota <- true
	}

	exCh := make(chan bool)
	go func() {
		defer func() {
			errI := recover()
			if errI != nil {
				err = fmt.Errorf("request handle fatal error: %v, TRACE: %v", errI, string(debug.Stack()))
				rsp := NewResponse(req.GetTransactionID(), RpcResponseStatusServiceProcessError, nil)
				rb := rsp.Encode()
				s.conn.SendData(rb)
			}
			<-s.reqProcQuota
			close(exCh)
		}()

		logger().Debug("accepted request",
			zap.String("name", req.GetServiceName()),
			zap.String("remoteAddr", s.conn.RemoteAddr().String()),
		)

		var hErr error
		for _, h := range s.cfg.OnRequest {
			if le := logger().Check(zap.DebugLevel, "calling OnRequest hook"); le != nil {
				le.Write(
					zap.String("hookName", reflect.TypeOf(h).Name()),
					zap.String("remoteAddr", s.conn.RemoteAddr().String()),
				)
			}
			subParam := new(protos.SubEvent)
			proto.UnmarshalMerge(req.GetMessageData(), subParam)
			hResponse, hErr := h.Fun.(OnRequest)(req, subParam, s, h.ExtraArgs...)
			if hErr != nil {
				logger().Warn("OnRequest hook err", zap.Error(hErr), zap.String("name", reflect.TypeOf(h).Name()))
				rsp := NewResponse(req.GetTransactionID(), RpcResponseStatusServiceProcessError, nil)
				rb := rsp.Encode()
				s.conn.SendData(rb)
				return
			} else if hResponse != nil {
				b, _ := proto.Marshal(hResponse)
				rsp := NewResponse(req.GetTransactionID(), RpcResponseStatusOk, b)
				_, hErr = s.conn.SendData(rsp.Encode())
				if hErr != nil {
					err = fmt.Errorf("failed to send OnRequest response data: %v", err)
				} else if le := logger().Check(zap.DebugLevel, "hook OnRequest processed."); le != nil {
					le.Write(
						zap.String("hookName", reflect.TypeOf(h).Name()),
						zap.String("remoteAddr", s.conn.RemoteAddr().String()),
					)
				}
				return
			}
		}

		// check service registration.
		shri, err := s.GetServiceHandlerInfo(req.GetServiceName())
		if err != nil {
			err = fmt.Errorf("service handler lookup err[%v]", err)
			_, errS := s.conn.SendData(NewResponse(req.GetTransactionID(), RpcResponseStatusServiceNotExists, nil).Encode())
			if errS != nil {
				err = fmt.Errorf("failed to send response[%v].---%v", errS, err)
			}
			return
		}

		paramType := reflect.New(shri.paramType).Interface().(proto.Message)
		err = proto.Unmarshal(req.GetMessageData(), paramType)
		if le := logger().Check(zap.DebugLevel, "processing request"); le != nil {
			le.Write(
				zap.String("name", req.GetServiceName()),
				zap.String("param", fmt.Sprintf("%#v", paramType)),
				zap.String("remoteAddr", s.conn.RemoteAddr().String()),
			)
		}

		if err != nil {
			logger().Warn(err.Error())
			rsp := NewResponse(req.GetTransactionID(), RpcResponseStatusParamIncorrect, nil)
			rb := rsp.Encode()
			s.conn.SendData(rb)
			return
		}

		result := shri.Handler(paramType)
		if le := logger().Check(zap.DebugLevel, "request processing complete"); le != nil {
			le.Write(
				zap.String("param", fmt.Sprintf("%#v", paramType)),
				zap.String("result", fmt.Sprintf("%#v", result)),
				zap.String("remoteAddr", s.conn.RemoteAddr().String()),
			)
		}

		b, err := proto.Marshal(result)
		if err != nil {
			err = fmt.Errorf("response data encoding err: %v", err)
			return
		}
		rsp := NewResponse(req.GetTransactionID(), RpcResponseStatusOk, b)
		rb := rsp.Encode()
		if le := logger().Check(zap.DebugLevel, "sending response data to client"); le != nil {
			le.Write(
				zap.String("name", req.GetServiceName()),
				zap.String("remoteAddr", s.conn.RemoteAddr().String()),
				zap.String("data", fmt.Sprintf("%v", rb)),
			)
		}

		_, err = s.conn.SendData(rb)
		if err != nil {
			err = fmt.Errorf("failed to send RPC response data: %v", err)
			return
		}

		for _, h := range s.cfg.AfterRequest {
			hErr = h.Fun.(AfterRequest)(req, result, s, err, h.ExtraArgs...)
			if hErr != nil {
				err = fmt.Errorf("AfterRequest hook err(%v)", hErr)
			}
		}
	}()

	if s.cfg.ReqProcessTimeout > 0 {
		select {
		case <-exCh:
			return
		case <-timer.C:
			err = RpcResponseStatusTimeout
		}
	} else {
		// no timeout
		<-exCh
	}
	return
}
